﻿using System;
using System.Collections.Generic;

using Apache.Cassandra;
using Client = Apache.Cassandra.Cassandra.Client;

using RexToy.Cassandra.Meta;

namespace RexToy.Cassandra
{
    class SuperColumnFamily<K, C, SUB> : ColumnFamilyBase<K, C>, ISuperColumnFamily<K, C, SUB>
    {
        public SuperColumnFamily(string keyspace, string columnFamily, ColumnFamilyInfo info, Client c, ConsistencyLevel read, ConsistencyLevel write)
            : base(keyspace, columnFamily, info, c, read, write)
        {
        }

        public void Insert(K key, C superColumn, object value)
        {
            throw ExceptionHelper.CreateUnsupportedOperation();
        }

        public void Insert(K key, C superColumn, SUB column, object value)
        {
            ColumnParent cp = Utils.CreateColumnParent(this.Name, this.GetColumnBytes(superColumn));

            var c = new Apache.Cassandra.Column();
            c.Name = this.GetSubColumnBytes(column);
            c.Timestamp = this.GetTimeStamp();
            c.Value = this.GetValueBytes(superColumn, value);

            _c.insert(this.GetKeyBytes(key), cp, c, _write);
        }

        public void Remove(K key)
        {
            ColumnPath cp = Utils.CreateColumnPath(this.Name, null, null);
            _c.remove(this.GetKeyBytes(key), cp, this.GetTimeStamp(), _write);
        }

        public void Remove(K key, C superColumn)
        {
            ColumnPath cp = Utils.CreateColumnPath(this.Name, this.GetColumnBytes(superColumn), null);
            _c.remove(this.GetKeyBytes(key), cp, this.GetTimeStamp(), _write);            
        }

        public void Remove(K key, C superColumn, SUB column)
        {
            ColumnPath cp = Utils.CreateColumnPath(this.Name, this.GetColumnBytes(superColumn), this.GetSubColumnBytes(column));
            _c.remove(this.GetKeyBytes(key), cp, this.GetTimeStamp(), _write);
        }

        public object Select(K key, C superColumn, SUB column)
        {
            ColumnPath cp = Utils.CreateColumnPath(this.Name, this.GetColumnBytes(superColumn), this.GetSubColumnBytes(column));

            var c = _c.get(this.GetKeyBytes(key), cp, _read);
            return this.GetValueData(superColumn, c.Column.Value);
        }

        public IRow<K, C> Select(K key, C superColumn)
        {
            var path = Utils.CreateColumnPath(this.Name, this.GetColumnBytes(superColumn), null);
            var sc = _c.get(this.GetKeyBytes(key), path, _read);
            var collection = this.CreateColumns(sc.Super_column.Columns);

            var super = new SuperColumn<C, SUB>(superColumn, collection);
            var scCollection = new SuperColumnCollection<C, SUB>();
            scCollection.AddColumn(super);
            return new Row<K, C>(key, scCollection);
        }

        public IRow<K, C> Select(K key, int count = 10, bool reversed = false)
        {
            return this._Select(key, default(C), default(C), count, reversed);
        }

        public IRow<K, C> Select(K key, C start, C finish, int count = 10, bool reversed = false)
        {
            return this._Select(key, start, finish, count, reversed);
        }

        private IRow<K, C> _Select(K key, C start, C finish, int count, bool reversed)
        {
            SlicePredicate predicate = Utils.CreateSlicePredicate(this.GetColumnBytes(start), this.GetColumnBytes(finish), count, reversed);
            ColumnParent cp = Utils.CreateColumnParent(this.Name);

            var list = _c.get_slice(this.GetKeyBytes(key), cp, predicate, _read);

            var collection = this.CreateColumns(list);
            return this.CreateRow(key, collection);
        }

        public IEnumerable<IRow<K, C>> Select(IEnumerable<K> keys, int count = 10, bool reversed = false)
        {
            return this._Select(keys, default(C), default(C), count, reversed);
        }

        public IEnumerable<IRow<K, C>> Select(IEnumerable<K> keys, C start, C finish, int count = 10, bool reversed = false)
        {
            return this._Select(keys, start, finish, count, reversed);
        }

        private IEnumerable<IRow<K, C>> _Select(IEnumerable<K> keys, C start, C finish, int count, bool reversed)
        {
            SlicePredicate predicate = Utils.CreateSlicePredicate(this.GetColumnBytes(start), this.GetColumnBytes(finish), count, reversed);
            ColumnParent cp = Utils.CreateColumnParent(this.Name);
            var data = _c.multiget_slice(this.ConvertKeyList(keys), cp, predicate, _read);

            var rows = new List<IRow<K, C>>();
            foreach (var kvp in data)
            {
                var scCollection = this.CreateColumns(kvp.Value);
                var row = this.CreateRow(this.GetKeyData(kvp.Key), scCollection);
                rows.Add(row);
            }
            return rows;
        }

        public IEnumerable<IRow<K, C>> List(int rowCount = 100, int columnCount = 10, bool reversedColumn = false)
        {
            return this._List(default(K), default(K), default(C), default(C), rowCount, columnCount, reversedColumn);
        }

        public IEnumerable<IRow<K, C>> List(C startColumn, C finishColumn, int rowCount = 100, int columnCount = 10, bool reversedColumn = false)
        {
            return this._List(default(K), default(K), startColumn, finishColumn, rowCount, columnCount, reversedColumn);
        }

        public IEnumerable<IRow<K, C>> List(K startKey, K finishKey, C startColumn, C finishColumn, int rowCount = 100, int columnCount = 10, bool reversedColumn = false)
        {
            return this._List(startKey, finishKey, startColumn, finishColumn, rowCount, columnCount, reversedColumn);
        }

        private IEnumerable<IRow<K, C>> _List(K startKey, K finishKey, C startColumn, C finishColumn, int rowCount, int columnCount, bool reversedColumn)
        {
            KeyRange kr = Utils.CreateKeyRange(this.GetKeyBytes(startKey), this.GetKeyBytes(finishKey), rowCount);
            SlicePredicate sp = Utils.CreateSlicePredicate(this.GetColumnBytes(startColumn), this.GetColumnBytes(finishColumn), columnCount, reversedColumn);
            ColumnParent cp = Utils.CreateColumnParent(this.Name);

            var result = _c.get_range_slices(cp, sp, kr, _read);

            var rowList = new List<IRow<K, C>>();
            foreach (var r in result)
            {
                var coll = this.CreateColumns(r.Columns);
                var row = this.CreateRow(this.GetKeyData(r.Key), coll);
                rowList.Add(row);
            }
            return rowList;
        }

        #region Abstract method implement
        protected ISuperColumnCollection<C, SUB> CreateColumns(List<ColumnOrSuperColumn> list)
        {
            SuperColumnCollection<C, SUB> scCollection = new SuperColumnCollection<C, SUB>();
            foreach (var sc in list)
            {
                ColumnCollection<SUB> collection = new ColumnCollection<SUB>();
                foreach (var c in sc.Super_column.Columns)
                {
                    collection.AddColumn(CreateColumn(c));
                }

                ISuperColumn<C, SUB> superCol = CreateColumn(sc.Super_column, collection);
                scCollection.AddColumn(superCol);
            }
            return scCollection;
        }

        protected IColumnCollection<SUB> CreateColumns(List<Apache.Cassandra.Column> list)
        {
            var collection = new ColumnCollection<SUB>();
            foreach (var col in list)
            {
                collection.AddColumn(this.CreateColumn(col));
            }
            return collection;
        }

        protected override IRow<K, C> CreateRow(K key, IColumnCollection<C> collection)
        {
            return new Row<K, C>(key, collection);
        }
        #endregion

        private IColumn<SUB> CreateColumn(Apache.Cassandra.Column column)
        {
            var col = this.GetSubColumnData(column.Name);
            return new Column<SUB>(col, this.GetValueData(col, column.Value));
        }

        private ISuperColumn<C, SUB> CreateColumn(Apache.Cassandra.SuperColumn column, ColumnCollection<SUB> collection)
        {
            var col = this.GetColumnData(column.Name);
            return new SuperColumn<C, SUB>(col, collection);
        }

        private byte[] GetSubColumnBytes(object column)
        {
            return _info.SubComparator.GetBytes(column);
        }

        private SUB GetSubColumnData(byte[] column)
        {
            return (SUB)_info.SubComparator.GetData(column);
        }

        object IColumnFamily<K, C>.Select(K key, C column)
        {
            return this.Select(key, column);
        }
    }
}
